package com.econ.powercloud.ems.service.biz;

import com.econ.powercloud.common.base.facade.util.JacksonUtil;
import com.econ.powercloud.console.facade.utils.CacheKeyUtils;
import com.econ.powercloud.ems.constant.MonitorConstant;
import com.econ.powercloud.ems.dal.entity.PrometheusMonitorCategory;
import com.econ.powercloud.ems.pojo.*;
import com.econ.powercloud.ems.service.dao.PrometheusMonitorCategoryDAO;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Multimap;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Service
public class PrometheusMonitorService {

    @Autowired
    private PrometheusMonitorCategoryDAO prometheusMonitorCategoryDAO;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    /**
     *
     * @return
     */
    public List<PrometheusNodeVO> getNodesInfo(){
        List<PrometheusMonitorCategory> categoryList = prometheusMonitorCategoryDAO.findByCategory(MonitorConstant.CATEGORY_NODE);
        List<PrometheusNodeVO> result = categoryList.stream().map(p -> {
            String ip = p.getIp();
            String monitorIp = p.getMonitorPort();
            Map<String, String> map = getCacheHashMap(ip, monitorIp, MonitorConstant.CATEGORY_NODE);
            PrometheusNodeVO node = JacksonUtil.convertToObj(map, PrometheusNodeVO.class);
            node.setIp(ip);
            node.setDescription(p.getDescription());
            return node;
        }).collect(Collectors.toList());
        return result;
    }

    public List<PrometheusRedisVO> getRedisInfo(){
        List<PrometheusMonitorCategory> categoryList = prometheusMonitorCategoryDAO.findByCategory(MonitorConstant.CATEGORY_REDIS);
        List<PrometheusRedisVO> result = categoryList.stream().map(p -> {
            String ip = p.getIp();
            String monitorIp = p.getMonitorPort();
            Map<String, String> map = getCacheHashMap(ip, monitorIp, MonitorConstant.CATEGORY_REDIS);
            PrometheusRedisVO node = JacksonUtil.convertToObj(map, PrometheusRedisVO.class);
            if(StringUtils.isNotBlank(p.getPort())){
                node.setPort(p.getPort());
            }
            node.setIp(ip);
            node.setDescription(p.getDescription());
            return node;
        }).collect(Collectors.toList());
        return result;
    }

    public List<PrometheusClickhouseVO> getClickhouseInfo(){
        List<PrometheusMonitorCategory> categoryList = prometheusMonitorCategoryDAO.findByCategory(MonitorConstant.CATEGORY_CLICKHOUSE);
        List<PrometheusClickhouseVO> result = categoryList.stream().map(p -> {
            String ip = p.getIp();
            String monitorIp = p.getMonitorPort();
            Map<String, String> map = getCacheHashMap(ip, monitorIp, MonitorConstant.CATEGORY_CLICKHOUSE);
            PrometheusClickhouseVO node = JacksonUtil.convertToObj(map, PrometheusClickhouseVO.class);
            if(StringUtils.isNotBlank(p.getPort())){
                node.setPort(p.getPort());
            }
            node.setIp(ip);
            node.setDescription(p.getDescription());
            return node;
        }).collect(Collectors.toList());
        return result;
    }

    public List<PrometheusMysqlVO> getMysqlInfo(){
        List<PrometheusMonitorCategory> categoryList = prometheusMonitorCategoryDAO.findByCategory(MonitorConstant.CATEGORY_MYSQL);
        List<PrometheusMysqlVO> result = categoryList.stream().map(p -> {
            String ip = p.getIp();
            String monitorIp = p.getMonitorPort();
            Map<String, String> map = getCacheHashMap(ip, monitorIp, MonitorConstant.CATEGORY_MYSQL);
            PrometheusMysqlVO node = JacksonUtil.convertToObj(map, PrometheusMysqlVO.class);
            if(StringUtils.isNotBlank(p.getPort())){
                node.setPort(p.getPort());
            }
            node.setIp(ip);
            node.setDescription(p.getDescription());
            return node;
        }).collect(Collectors.toList());
        return result;
    }

    public List<PrometheusKafkaVO> getKafkaInfo(){

        List<PrometheusMonitorCategory> topicList = prometheusMonitorCategoryDAO.findByCategory(MonitorConstant.CATEGORY_KAFKA_TOPIC);
        Map<String, List<PrometheusKafkaTopicVO>> topicMap = new HashMap<>();
        for(PrometheusMonitorCategory category : topicList){
            Map<String, Map<String,String>> multiMap = new HashMap<>();
            Map<String, String> map = getCacheHashMap(category.getIp(), category.getMonitorPort(), MonitorConstant.CATEGORY_KAFKA_TOPIC);

            Iterator iterator = map.entrySet().iterator();
            while (iterator.hasNext()){
                Map.Entry<String,String> entry = (Map.Entry) iterator.next();
                String key = entry.getKey();
                String value = entry.getValue();
                String[] keys = key.split("#");
                String topic = keys[1];
                String item = keys[0];
                if(multiMap.containsKey(topic)){
                    multiMap.get(topic).put(item,value);
                } else {
                    Map<String, String> m = new HashMap<>();
                    m.put(item, value);
                    multiMap.put(topic, m);
                }
            }
            List<PrometheusKafkaTopicVO> topicVOList = Lists.newArrayList();
            Iterator iterator1 = multiMap.entrySet().iterator();
            while (iterator1.hasNext()){
                Map.Entry<String,Map<String,String>> entry = (Map.Entry) iterator1.next();
                PrometheusKafkaTopicVO topicVO = JacksonUtil.convertToObj(entry.getValue(), PrometheusKafkaTopicVO.class);
                topicVO.setTopic(entry.getKey());
                topicVOList.add(topicVO);
            }
            topicMap.put(category.getIp(), topicVOList);
        }

        List<PrometheusMonitorCategory> categoryList = prometheusMonitorCategoryDAO.findByCategory(MonitorConstant.CATEGORY_KAFKA);
        List<PrometheusKafkaVO> result = categoryList.stream().map(p -> {
            String ip = p.getIp();
            String monitorIp = p.getMonitorPort();
            Map<String, String> map = getCacheHashMap(ip, monitorIp, MonitorConstant.CATEGORY_KAFKA);
            PrometheusKafkaVO node = JacksonUtil.convertToObj(map, PrometheusKafkaVO.class);
            if(StringUtils.isNotBlank(p.getPort())){
                node.setPort(p.getPort());
            }
            if(topicMap.containsKey(ip)){
                node.setTopics(topicMap.get(ip));
            }
            node.setIp(ip);
            node.setDescription(p.getDescription());
            return node;
        }).collect(Collectors.toList());
        return result;
    }

    private Map<String, String> getCacheHashMap(String ip, String monitorPort, String category){
        String instance = ip + "-" + monitorPort;
        String redisKey = CacheKeyUtils.getMonitorDataKey(instance, category);
        HashOperations<String, String, String> hashOperations = stringRedisTemplate.opsForHash();
        Map<String, String> result = hashOperations.entries(redisKey);
        return result;
    }
}
